package com.shanks.framework.mq.rocketmq.producer;

import com.shanks.framework.mq.common.constant.MQConstant;
import com.shanks.framework.mq.exception.MQException;
import com.shanks.framework.mq.producer.AbstractMQProducer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RocketMQProducerImpl extends AbstractMQProducer {
    private static final Logger logger = LoggerFactory.getLogger(RocketMQProducerImpl.class);

    private DefaultMQProducer defaultMQProducer;
    private RocketMQProducerAttribute rocketMQProducerAttribute;

    public RocketMQProducerImpl(DefaultMQProducer defaultMQProducer, RocketMQProducerAttribute rocketMQProducerAttribute) {
        this.defaultMQProducer = defaultMQProducer;
        this.rocketMQProducerAttribute = rocketMQProducerAttribute;
    }

    @Override
    protected void doSend(String topic, String tags, String key, byte[] body, String source, String sign) throws MQException {
        try {
            Message msg = new Message(topic, tags, key, body);
            if (source != null) {
                msg.putUserProperty(MQConstant.MSG.MSG_SOURCE, source);
            }
            if (sign != null) {
                msg.putUserProperty(MQConstant.MSG.PROPERTY_SIGNATURE, sign);
            }
            SendResult sendResult = defaultMQProducer.send(msg);
            if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                logger.error("发送消息结果异常: {}", sendResult);
                throw new MQException("发送消息结果异常：" + sendResult.getSendStatus());
            }
            // 检查发送结果
            if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                logger.error("发送消息结果异常: {}", sendResult);
                throw new MQException("发送消息失败，结果状态：" + sendResult.getSendStatus());
            }

            // 记录成功日志（可选）
            logger.info("消息发送成功: topic={}, tags={}, key={}, sendResult={}", topic, tags, key, sendResult);

        } catch (MQException e) {
            // 捕获自定义异常直接抛出
            logger.error("自定义MQ异常，消息发送失败: topic={}, tags={}, key={}, error={}", topic, tags, key, e.getMessage(), e);
            throw e;
        } catch (Exception e) {
            // 捕获其他异常，包装为 MQException
            logger.error("发送消息失败: topic={}, tags={}, key={}, error={}", topic, tags, key, e.getMessage(), e);
            throw new MQException("发送消息失败", e);
        }
    }

    @Override
    public String getNamesrvAddr() {
        return defaultMQProducer.getNamesrvAddr();
    }
}
